package com.qsl.hook;

import com.qsl.hook.exception.HookException;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Objects;

/**
 * @description: DataX 管理器
 * @author: 青石路
 * @date: 2024/4/22 14:42
 */

public class DataXManager {

    private static final String SYSTEM_ENCODING = System.getProperty("sun.jnu.encoding");
    private static final Logger LOGGER = LoggerFactory.getLogger(DataXManager.class);

    /**
     * command = java -server -Xms1g -Xmx1g -XX:+HeapDumpOnOutOfMemoryError -Ddatax.home=/datax
     *  -classpath /datax/lib/* com.alibaba.datax.core.Engine -mode standalone -core %s -job %s
     * @param command 启动DataX的命令
     * @param coreContent core.json内容，而非路径；可为空
     * @param jobContent job.json内容，而非路径；不可为空
     */
    public void exec(String command, String coreContent, String jobContent) {

        if (StringUtils.isBlank(command)) {
            throw new HookException("DataX 启动命令为空");
        }

        if (StringUtils.isBlank(jobContent)) {
            throw new HookException("job配置为空");
        }

        Path coreJsonPath = null;
        Path jobJsonPath = null;
        try {
            // json 先写入临时文件
            if (StringUtils.isNoneBlank(coreContent)) {
                coreJsonPath = Files.createTempFile("core", ".json");
                Files.write(coreJsonPath, coreContent.getBytes());
            }
            jobJsonPath = Files.createTempFile("job", ".json");
            Files.write(jobJsonPath, jobContent.getBytes());
            LOGGER.info("coreJsonPath={}, jobJsonPath={}", coreJsonPath, jobJsonPath);
            String corePath = Objects.isNull(coreJsonPath) ? "" : coreJsonPath.toString();
            String realCommand = String.format(command, corePath, jobJsonPath);
            LOGGER.info("command = {}", realCommand);

            Process process = Runtime.getRuntime().exec(realCommand);
            // TODO 可以考虑用线程池
            new Thread(() -> {
                try (BufferedReader reader = new BufferedReader(
                        new InputStreamReader(process.getInputStream(), SYSTEM_ENCODING))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.contains("sync rows=")) {
                            long syncRows = Long.parseLong(line.split("=")[1]);
                            // TODO 实际项目中应该进行 已同步数据量 持久化，类似：update table_name set sync_rows = sync_rows + syncRows
                        } else {
                            LOGGER.info(line);
                        }
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }).start();

            new Thread(() -> {
                try (BufferedReader errorReader = new BufferedReader(
                        new InputStreamReader(process.getErrorStream(), SYSTEM_ENCODING))) {
                    String line;
                    while ((line = errorReader.readLine()) != null) {
                        if (line.contains("sync rows=")) {
                            long syncRows = Long.parseLong(line.split("=")[1]);
                            // TODO 实际项目中应该进行 已同步数据量 持久化，类似：update table_name set sync_rows = sync_rows + syncRows
                        } else {
                            LOGGER.error(line);
                        }
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }).start();

            // 等待命令执行完成
            int i = process.waitFor();
            if (i == 0) {
                LOGGER.info("DataX 执行完成");
            } else {
                LOGGER.error("DataX 执行失败");
            }
        } catch (Exception e) {
            LOGGER.error("job执行异常：", e);
        } finally {
            try {
                if (Objects.nonNull(coreJsonPath)) {
                    Files.deleteIfExists(coreJsonPath);
                }
                if (Objects.nonNull(jobJsonPath)) {
                    Files.deleteIfExists(jobJsonPath);
                }
            } catch (IOException ex) {
                LOGGER.error("临时文件删除失败：", ex);
            }
        }
    }
}
